package com.ibm.cps.sql.window;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.sql.AggregationDescriptor;
import com.ibm.cps.sql.AggregationDescriptor.WindowType;
import com.ibm.cps.sql.ProcessorMetadataCreator;
import com.ibm.cps.sql.ProjectionParser;
import com.ibm.cps.sql.ProjectionParser.DatasourceField;
import com.ibm.cps.sql.SelectItemDescriptor;
import com.ibm.cps.sql.expression.BooleanExpression;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.window.WindowProcessorCreator.ProcessorDescriptor;
import com.ibm.factories.DataSourceFactory;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.SchemaCache;
import com.ibm.util.exception.CPSException;

/**
 * Created by telekinesis on 6/5/15.
 */
public class WindowingRequestParser {
	private final DataSourceFactory datasourceFactory;
	private final SchemaCache schemaCache;
	private final String ruleName;
	private final String groupId;
	private final String tenantId;
	private final ProcessorMetadataCreator metaCreator;

	public WindowingRequestParser(DataSourceFactory datasourceFactory,
			SchemaCache schemaCache, String ruleName, String groupId,
			String tenantId) {
		this.datasourceFactory = datasourceFactory;
		this.schemaCache = schemaCache;
		this.ruleName = ruleName;
		this.groupId = groupId;
		this.tenantId = tenantId;
		this.metaCreator = new ProcessorMetadataCreator(tenantId, groupId, "sql");
	}

	// select dft(pressure) as RULENAME_1 from topic
	// group by SlidingWindow(10)
	// having RULENAME_1(0) > 100 and RULENAME_1(0) < 200
	public ArrayNode parseWindowingRequest(
			List<String> dataSources, SelectItemDescriptor outputFields, AggregationDescriptor aggregation, BooleanExpression filteringCondition,
			List<String> publishers) throws CPSException {
		ArrayNode result = JSONUtility.newArrayNode();
		ObjectNode inputProcessor = createInputProcessors(dataSources, groupId, tenantId);
		String lastProcessorId = inputProcessor.get(MessageFields.PROCESSORID).asText();
		result.add(inputProcessor);
		ObjectNode projectionNode = createInitialProjection(dataSources, outputFields, aggregation, filteringCondition, lastProcessorId);
		result.add(projectionNode);
		lastProcessorId = projectionNode.get(MessageFields.PROCESSORID).asText();

		ObjectNode windowNode = createWindowNode(aggregation, lastProcessorId);
		lastProcessorId = windowNode.get(MessageFields.PROCESSORID).asText();
		result.add(windowNode);
		
		ArrayNode calculationAndFiltering = createCalculatingProcessors(lastProcessorId, outputFields, aggregation, filteringCondition);
		result.addAll(calculationAndFiltering);
		lastProcessorId = calculationAndFiltering.get(calculationAndFiltering.size() - 1).get(MessageFields.PROCESSORID).asText();

		ArrayNode publisherNodes = createPublishers(publishers, lastProcessorId);
		result.addAll(publisherNodes);		
		return result;
	}
	
	private ObjectNode createInputProcessors(List<String> dataSources, String groupId, String tenantId) throws CPSException{
		// TODO: support multiple datasource, that should be a join
		if(dataSources.size() > 1){
			throw new CPSException(ErrorCode.INPUT_ERROR, "Selecting from multiple datasources is not supported by now");
		}
		ObjectNode inputProcessor = metaCreator.createInputProcessor(dataSources.get(0));
		return inputProcessor;
	}
	
	private ObjectNode createInitialProjection(List<String> datasources, SelectItemDescriptor projection, AggregationDescriptor aggregation, BooleanExpression filterCondition, String parentProcessorId) throws CPSException{
		ProjectionParser parser = new ProjectionParser(tenantId, datasourceFactory, schemaCache);
		Set<DatasourceField> fields = parser.parseProjection(datasources, projection, aggregation, filterCondition);
		Map<String, String> fieldMapping = new HashMap<>();
		for (DatasourceField f : fields) {
			String fullName = f.fullQualifiedName();
			fieldMapping.put(fullName, fullName);
		}
		ObjectNode projectionNode = metaCreator.createProjectionWithoutCalculation(fieldMapping, parentProcessorId);
		return projectionNode;
	}
	
	private ObjectNode createWindowNode(AggregationDescriptor aggregation, String parentProcessorId){
		WindowDescriptor window = aggregation.getWindow();
		WindowType type = window.getType();
		long step = window.getInterval();
		long size = window.getSize();
		switch(type){
			case TIME:
				return metaCreator.createTimeWindowNode(parentProcessorId, step, size);
			default:
				return metaCreator.createCountWindowNode(parentProcessorId, step, size);
		}
	}
	
	private ArrayNode createCalculatingProcessors(String parentProcessorId, SelectItemDescriptor outputFields, AggregationDescriptor aggregation, BooleanExpression filteringCondition) throws CPSException{
		//Currently aggregation is not concerned
		ArrayNode result = JSONUtility.newArrayNode();
		WindowProcessorCreator creator = new WindowProcessorCreator(tenantId, groupId, ruleName);
		Map<String, ValueExpression> selectItems = outputFields.getColumnExpression();
		for (String alias : selectItems.keySet()) {
			ValueExpression expression = selectItems.get(alias);
			ProcessorDescriptor itemProcessors = creator.create(parentProcessorId, expression, alias);
			if (itemProcessors.processors.size() != 1) {
				parentProcessorId = itemProcessors.lastProcessorId;
			}
			result.addAll(itemProcessors.processors);
			parentProcessorId = itemProcessors.lastProcessorId;
		}
		Map<String, String> fieldNameMapping = new HashMap<>();
		for (String outputFieldName : selectItems.keySet()) {
			fieldNameMapping.put(outputFieldName, outputFieldName);
		}
		Map<String, String> conditionFields = collectAggregationAndFilterFields(aggregation, filteringCondition);
		fieldNameMapping.putAll(conditionFields);
		ObjectNode finalProjection = metaCreator.createProjectionWithoutCalculation(fieldNameMapping, parentProcessorId);
		result.add(finalProjection);
		parentProcessorId = finalProjection.get(MessageFields.PROCESSORID).asText();
		ArrayNode processorsForCondition = creator.createProcessorsForCondition(parentProcessorId, filteringCondition);
		result.addAll(processorsForCondition);
		return result;
	}
	
	//In pre-processing, all field names are replaced by full qualified name <datasource_name>.<field_name>
	private Map<String, String> collectAggregationAndFilterFields(AggregationDescriptor aggregation, BooleanExpression filterCondition){
		Map<String, String> fields = new HashMap<>();
		Set<String> aggregationKeys = aggregation.getAggregationKeyFields();
		for (String key : aggregationKeys) {
			fields.put(key, key);
		}
		Set<String> filterFields = filterCondition.collectVariableNames();
		for (String f : filterFields) {
			fields.put(f, f);
		}
		return fields;
	}
	
	private ArrayNode createPublishers(List<String> publisherNames, String parentProcessorId){
		ArrayNode nodes = JSONUtility.newArrayNode();
		for (String p : publisherNames) {
			ObjectNode publisherNode = metaCreator.createPublishNode(p, parentProcessorId);
			nodes.add(publisherNode);
		}
		return nodes;
	}
	
}
